 1.Kafka高级特性解析之分区中的副本机制
   
   Kafka在一定数量的服务器上对主题分区进行复制。
   当集群中的一个broker宕机后系统可以⾃动故障转移到其他可用的副本上，不会造成数据丢失。
   --replication-factor 3 1leader+2follower
   1).将复制因为为1的未复制主题称为复制主题。
   2).主题的分区是复制的最小单元。
   3).在非故障情况下，Kafka中的每个分区都有一个Leader副本和零个或多个Follower副本。
   4).包括Leader副本在内的副本总数构成复制因子。
   5).所有读取和写入都由Leader副本负责。
   6).通常，分区比broker多，并且Leader分区在broker之间平均分配。
   Follower分区像普通的Kafka消费者一样，消费来自Leader分区的消息，并将其持久化到自己的日志中。
   允许Follower对日志条目拉取进行批处理。
   同步节点定义：
   1).节点必须能够维持与ZooKeeper的会话（通过ZooKeeper的⼼跳机制）
   2).对于Follower副本分区，它复制在Leader分区上的写入，并且不要延迟太多
   Kafka提供的保证是，只要有至少一个同步副本处于活动状态，提交的消息就不会丢失。
   宕机如何恢复
   (1).少部分副本宕机
   (2).全部副本宕机
   当全部副本宕机了有两种恢复方式
   1).等待ISR中的一个恢复后，并选它作为leader。（等待时间较长，降低可用性）
   2).选择第一个恢复的副本作为新的leader，无论是否在ISR中。(并未包含之前leader commit的数据，因此造成
数据丢失)
 
 2.Leader选举
   
   下图中
   分区P1的Leader是0，ISR是0和1
   分区P2的Leader是2，ISR是1和2
   分区P3的Leader是1，ISR是0，1，2。
   生产者和消费者的请求都由Leader副本来处理。Follower副本只负责消费Leader副本的数据和Leader保持同步。
   对于P1，如果0宕机会发生什么？
   Leader副本和Follower副本之间的关系并不是固定不变的,在Leader所在的broker发⽣故障的时候，就需要进行
分区的Leader副本和Follower副本之间的切换，需要选举Leader副本。
   如何选举？
   如果某个分区所在的服务器除了问题，不可用，kafka会从该分区的其他的副本中选择一个作为新的Leader。之后
所有的读写就会转移到这个新的Leader上。现在的问题是应当选择哪个作为新的Leader。
   只有那些跟Leader保持同步的Follower才应该被选作新的Leader。
   Kafka会在Zookeeper上针对每个Topic维护一个称为ISR（in-sync replica，已同步的副本）的集合，该集合中是
一些分区的副本。
   只有当这些副本都跟Leader中的副本同步了之后，kafka才会认为消息已提交，并反馈给消息的生产者。
   如果这个集合有增减，kafka会更新zookeeper上的记录。
   如果某个分区的Leader不可用，Kafka就会从ISR集合中选择一个副本作为新的Leader。
   显然通过ISR，kafka需要的冗余度较低，可以容忍的失败数比较高。
   假设某个topic有N+1 个副本，kafka可以容忍N个服务器不可用。
   为什么不用少数服从多数的方法
   少数服从多数是一种比较常见的一致性算发和Leader选举法。
   它的含义是只有超过半数的副本同步了，系统才会认为数据已同步；
   选择Leader时也是从超过半数的同步的副本中选择。
   这种算法需要较高的冗余度，跟Kafka比起来，浪费资源。
   譬如只允许一台机器失败，需要有三个副本；而如果只容忍两台机器失败，则需要五个副本。
   而kafka的ISR集合方法，分别只需要两个和三个副本。
   如果所有的ISR副本都失败了怎么办？
   此时有两种方法可选，
   1).等待ISR集合中的副本复活，
   2).选择任何一个立即可用的副本，而这个副本不一定是在ISR集合中。
              需要设置unclean.leader.election.enable=true
   这两种方法各有利弊，实际生产中按需选择。
   如果要等待ISR副本复活，虽然可以保证一致性，但可能需要很长时间。而如果选择立即可用的副本，则很可能该
副本并不一致。
   总结：
   Kafka中Leader分区选举，通过维护一个动态变化的ISR集合来实现，一旦Leader分区丢掉，则从ISR中随机挑选
一个副本做新的Leader分区。
   如果ISR中的副本都丢失了，则：
   1).可以等待ISR中的副本任何一个恢复，接着对外提供服务，需要时间等待。
   2).从OSR中选出一个副本做Leader副本，此时会造成数据丢失
 
 3.分区重新分配
   
   向已经部署好的Kafka集群里面添加机器，我们需要从已经部署好的Kafka节点中复制相应的配置文件，然后把里
面的broker id修改成全局唯一的，最后启动这个节点即可将它加入到现有Kafka集群中。
   问题：新添加的Kafka节点并不会自动地分配数据，无法分担集群的负载，除非我们新建一个topic。
   需要手动将部分分区移到新添加的Kafka节点上，Kafka内部提供了相关的工具来重新分布某个topic的分区。
   在重新分布topic分区之前，我们先来看看现在topic的各个分区的分布位置：
   1).创建主题
   [root@linux121 ~]# kafka-topics.sh --zookeeper linux121:2181/myKafka --create --topic
tp_re_01 --partitions 5 --replication-factor 1
   2).查看主题信息
   [root@linux121 ~]# kafka-topics.sh --zookeeper linux121:2181/myKafka --describe --
topic tp_re_01
   Topic:tp_re_01 PartitionCount:5 ReplicationFactor:1 Configs:
   Topic: tp_re_01 Partition: 0 Leader: 0 Replicas: 0 Isr: 0
   Topic: tp_re_01 Partition: 1 Leader: 0 Replicas: 0 Isr: 0
   Topic: tp_re_01 Partition: 2 Leader: 0 Replicas: 0 Isr: 0
   Topic: tp_re_01 Partition: 3 Leader: 0 Replicas: 0 Isr: 0
   Topic: tp_re_01 Partition: 4 Leader: 0 Replicas: 0 Isr: 0
   3).在linux122搭建kafka
        拷贝jdk并安装
   [root@linux121 opt]# scp jdk-8u261-linux-x64.rpm linux122:~
   此处不需要zookeeper，切记！！！
   让配置生效：
   . /etc/profile
       拷贝linux121上安装的Kafka
   [root@linux121 opt]# scp -r kafka_2.12-1.0.2/ linux122:/opt
   修改linux122上Kafka的配置：
   broker.id=1
   zookeeper.coonect=linux121:2181/myKafka
   启动Kafka
   [root@linux122 ~]# kafka-server-start.sh /opt/kafka_2.12-1.0.2/config/server.properties
   注意观察linux122上节点启动的时候的ClusterId，看和zookeeper节点上的ClusterId是否一致，如果是，证
明linux122和linux121在同一个集群中。
   linux122启动的Cluster ID：
   zookeeper节点上的Cluster ID：
   在linux121上查看zookeeper的节点信息：
   4).现在我们在现有集群的基础上再添加一个Kafka节点，然后使用Kafka自带的kafka-reassign-partitions.
sh 工具来重新分布分区。该工具有三种使用模式：
   (1).generate模式，给定需要重新分配的Topic，自动生成reassign plan（并不执行）
   (2).execute模式，根据指定的reassign plan重新分配Partition
   (3).verify模式，验证重新分配Partition是否成功
   5).我们将分区3和4重新分布到broker1上，借助kafka-reassign-partitions.sh工具生成reassign plan，不过我
们先得按照要求定义一个文件，里面说明哪些topic需要重新分区，文件内容如下：
   [root@linux121 ~]# cat topics-to-move.json
   {
     "topics": [
        {
         "topic":"tp_re_01"
        }
     ],
     "version":1
   }
   [root@linux121 ~]# kafka-reassign-partitions.sh --zookeeper linux121:2181/myKafka 
--topics-to-move-json-file topics-to-move.json --broker-list "0,1" --generate
Current partition replica assignment
{"version":1,"partitions":[{"topic":"tp_re_01","partition":4,"replicas":[0],"log_dirs":
["any"]},{"topic":"tp_re_01","partition":1,"replicas":[0],"log_dirs":["any"]},
{"topic":"tp_re_01","partition":2,"replicas":[0],"log_dirs":["any"]},
{"topic":"tp_re_01","partition":3,"replicas":[0],"log_dirs":["any"]},
{"topic":"tp_re_01","partition":0,"replicas":[0],"log_dirs":["any"]}]}
Proposed partition reassignment configuration
{"version":1,"partitions":[{"topic":"tp_re_01","partition":4,"replicas":[0],"log_dirs":
["any"]},{"topic":"tp_re_01","partition":1,"replicas":[1],"log_dirs":["any"]},
{"topic":"tp_re_01","partition":2,"replicas":[0],"log_dirs":["any"]},
{"topic":"tp_re_01","partition":3,"replicas":[1],"log_dirs":["any"]},
{"topic":"tp_re_01","partition":0,"replicas":[0],"log_dirs":["any"]}]}
   Proposed partition reassignment configuration下面生成的就是将分区重新分布到broker 1 上的结果。我们将
这些内容保存到名为result.json文件里面（文件名不重要，文件格式也不一定要以json为结尾，只要保证内容是json
即可)，然后执行这些reassign plan：
   执行计划：
   [root@linux121 ~]# kafka-reassign-partitions.sh --zookeeper linux121:2181/myKafka 
--reassignment-json-file my-topics-to-move-execute.json --execute
Current partition replica assignment
{"version":1,"partitions":[{"topic":"tp_re_01","partition":4,"replicas":[0],"log_dirs":
["any"]},{"topic":"tp_re_01","partition":1,"replicas":[0],"log_dirs":["any"]},
{"topic":"tp_re_01","partition":2,"replicas":[0],"log_dirs":["any"]},
{"topic":"tp_re_01","partition":3,"replicas":[0],"log_dirs":["any"]},
{"topic":"tp_re_01","partition":0,"replicas":[0],"log_dirs":["any"]}]}
Save this to use as the --reassignment-json-file option during rollback
Successfully started reassignment of partitions.
   [root@linux121 ~]#
   这样Kafka就在执行reassign plan，我们可以校验reassign plan是否执行完成：
   [root@linux121 ~]# kafka-reassign-partitions.sh --zookeeper linux121:2181/myKafka --
reassignment-json-file my-topics-to-move-execute.json --verify
Status of partition reassignment:
Reassignment of partition tp_re_01-1 completed successfully
Reassignment of partition tp_re_01-4 completed successfully
Reassignment of partition tp_re_01-2 completed successfully
Reassignment of partition tp_re_01-3 completed successfully
Reassignment of partition tp_re_01-0 completed successfully
  [root@linux121 ~]#
  查看主题的细节：
  分区的分布的确和操作之前不一样了，broker 1 上已经有分区分布上去了。使用kafka-reassign-partitions.
sh 工具生成的reassign plan只是一个建议，方便大家而已。其实我们自己完全可以编辑一个reassign
plan，然后执行它，如下：
  {
    "version": 1,
    "partitions": [{
      "topic": "tp_re_01",
      "partition": 4,
      "replicas": [1],
      "log_dirs": ["any"]
  }, {
      "topic": "tp_re_01",
      "partition": 1,
      "replicas": [0],
      "log_dirs": ["any"]
  }, {
      "topic": "tp_re_01",
      "partition": 2,
      "replicas": [0],
      "log_dirs": ["any"]
  }, {
      "topic": "tp_re_01",
      "partition": 3,
      "replicas": [1],
      "log_dirs": ["any"]
  }, {
	  "topic": "tp_re_01",
      "partition": 0,
      "replicas": [0],
      "log_dirs": ["any"]
  }]
}
  将上面的json数据文件保存到my-topics-to-execute.json文件中，然后也是执行它：
  [root@linux121 ~]# kafka-reassign-partitions.sh --zookeeper linux121:2181/myKafka --
reassignment-json-file my-topics-to-execute.json --execute
Current partition replica assignment
{"version":1,"partitions":[{"topic":"tp_re_01","partition":4,"replicas":
[0],"log_dirs":["any"]},{"topic":"tp_re_01","partition":1,"replicas":[1],"log_dirs":
["any"]},{"topic":"tp_re_01","partition":2,"replicas":[0],"log_dirs":["any"]},
{"topic":"tp_re_01","partition":3,"replicas":[1],"log_dirs":["any"]},
{"topic":"tp_re_01","partition":0,"replicas":[0],"log_dirs":["any"]}]}
Save this to use as the --reassignment-json-file option during rollback
Successfully started reassignment of partitions.
  [root@linux121 ~]# kafka-reassign-partitions.sh --zookeeper linux121:2181/myKafka --
reassignment-json-file my-topics-to-execute.json --verify
Status of partition reassignment:
Reassignment of partition tp_re_01-1 completed successfully
Reassignment of partition tp_re_01-4 completed successfully
Reassignment of partition tp_re_01-2 completed successfully
Reassignment of partition tp_re_01-3 completed successfully
Reassignment of partition tp_re_01-0 completed successfully
  等这个reassign plan执行完，我们再来看看分区的分布：
  [root@linux121 ~]# kafka-topics.sh --zookeeper linux121:2181/myKafka --describe 
--topic tp_re_01
Topic:tp_re_01 PartitionCount:5 ReplicationFactor:1 Configs:
Topic: tp_re_01 Partition: 0 Leader: 0 Replicas: 0 Isr: 0
Topic: tp_re_01 Partition: 1 Leader: 0 Replicas: 0 Isr: 0
Topic: tp_re_01 Partition: 2 Leader: 0 Replicas: 0 Isr: 0
Topic: tp_re_01 Partition: 3 Leader: 1 Replicas: 1 Isr: 1
Topic: tp_re_01 Partition: 4 Leader: 1 Replicas: 1 Isr: 1
  [root@linux121 ~]# 
  搞定！
  
   